﻿// Copyright (c) Microsoft Open Technologies, Inc. All rights reserved. 
// See License.txt in the project root for license information.
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics.Contracts;
using System.Diagnostics.Tracing;
using System.Linq;
using System.Text;
using System.Threading.Actors.Utilities;
using System.Threading.Tasks;

namespace System.Threading.Actors.StatelessActor
{
    // Class that coordinates all IsolatedActorStates that compose the aggregate state of this service.
    internal class NonReplicatingStateCoordinator : IPersistenceHandler, IDisposable
    {
        private readonly StatelessActorStateTypes m_storageType;
        private readonly string m_serviceName;
        private readonly StateBufferingHelper m_stateHelper;
        private readonly IsolatedActorState m_defaultState;
        private ConcurrentDictionary<string, IsolatedActorState> m_keySpecificStates = new ConcurrentDictionary<string, IsolatedActorState>();
        private static readonly Task s_preCompletedTask = Task.FromResult(true);
        private readonly ActorETWEventListener m_ETWListener;

        public NonReplicatingStateCoordinator(string serviceName, StatelessActorStateTypes storageType)
        {
            m_serviceName = serviceName;
            m_storageType = storageType;
            m_stateHelper = new StateBufferingHelper(this);

            // Kick off real-time logging of Actor ETW events
            m_ETWListener = new ActorETWEventListener(serviceName);
            // Use IActor.Configure to change this at runtime.
            m_ETWListener.EnableEvents(ActorETWEventSource.Log, EventLevel.Informational);

            var actorInteractor = new FabricActorInteractionHelper();
            m_defaultState = new IsolatedActorState(m_stateHelper, "", serviceName, m_ETWListener, actorInteractor); 
        }

        // Retrieve actor state associated with partitionKey, creating it if necessary
        public IsolatedActorState GetActorState(string partitionKey)
        {
            Contract.Ensures(Contract.Result<IsolatedActorState>() != null);

            if(String.IsNullOrEmpty(partitionKey)) return m_defaultState;

            IsolatedActorState result;
            if (!m_keySpecificStates.TryGetValue(partitionKey, out result))
            {
                lock (m_keySpecificStates)
                {
                    if (!m_keySpecificStates.TryGetValue(partitionKey, out result))
                    {
                        var actorInteractor = new FabricActorInteractionHelper();
                        result = new IsolatedActorState(m_stateHelper, partitionKey, m_serviceName, m_ETWListener, actorInteractor);
                        m_keySpecificStates[partitionKey] = result;
                    }
                }
            }
            else
            {
                Contract.Assert(result != null, "Cached IsolatedActorState for a partition was null!");
            }

            return result;
        }
        
        // We've aggregated a number of changes to persist.
        public Task OnPersistAsync(IList<ChangeOperation> toPersist)
        {
            if (m_storageType == StatelessActorStateTypes.InMemory)
            {
                return PersistToMemory(toPersist);
            }
            else
            {
                throw new InvalidOperationException("Unknown stateType = " + m_storageType);
            }
        }

        // Support for "InMemory" state type
        private Task PersistToMemory(IList<ChangeOperation> toPersist)
        {
            foreach (var change in toPersist)
            {
                string partitionName = change.PartitionName;
                IsolatedActorState state = GetActorState(partitionName);
                if (change.OperationType == ChangeOperationType.Set)
                {
                    state.SetDurableValue(change.OperationKey, change.OperationValue, "OnPersistAsync", scour: true);
                }
                else if (change.OperationType == ChangeOperationType.Remove)
                {
                    state.RemoveDurableValue(change.OperationKey);
                }
                else throw new InvalidOperationException("Unexpected OperationType " + change.OperationType);
            }

            return s_preCompletedTask;
        }

        // Clean up
        public void Dispose()
        {
            m_stateHelper.Dispose();
            m_keySpecificStates.Clear();
            if (m_ETWListener != null) m_ETWListener.Dispose();
        }
    }
}
